package com.atguigu.edu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.bean.TrafficVistorBean;
import com.atguigu.edu.util.ClickHouseUtil;
import com.atguigu.edu.util.DateFormatUtil;
import com.atguigu.edu.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import java.text.DecimalFormat;
import java.time.Duration;

public class DwsTrafficVisitorWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //TODO　2 设置状态后端
                 /*
                 env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
                 env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
                 env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
                 env.setStateBackend(new HashMapStateBackend());
                 env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
                 System.setProperty("HADOOP_USER_NAME", "atguigu");
                  */

        //TODO 3 读取kafka  dwd独立访客主题的数据
        String uvTopic = "dwd_traffic_unique_visitor_detail";
        String pageTopic = "dwd_traffic_page_log";
        String ujTopic = "dwd_traffic_user_jump_detail";
        String groupId = "dwsTrafficVisitorWindow";
        DataStreamSource<String> uvStream = env.addSource(KafkaUtil.getKafkaConsumer(uvTopic, groupId));
        DataStreamSource<String> pageStream = env.addSource(KafkaUtil.getKafkaConsumer(pageTopic, groupId));
        DataStreamSource<String> ujStream = env.addSource(KafkaUtil.getKafkaConsumer(ujTopic, groupId));

        //TODO 4 转换格式
        SingleOutputStreamOperator<TrafficVistorBean> uvBeanStream =
                uvStream.flatMap(new FlatMapFunction<String, TrafficVistorBean>() {

                    @Override
                    public void flatMap(String s, Collector<TrafficVistorBean> collector) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(s);
                        JSONObject common = jsonObject.getJSONObject("common");
                        collector.collect(TrafficVistorBean.builder()
                                .isNew(common.getString("is_new"))
                                .uvCt(1L)
                                .svCt(0L)
                                .pvCt(0L)
                                .ujCt(0L)
                                .durSum(0L)
                                .ts(jsonObject.getLong("ts"))
                                .build());
                    }
                });
        SingleOutputStreamOperator<TrafficVistorBean> pageBeanStream =
                pageStream.flatMap(new FlatMapFunction<String, TrafficVistorBean>() {
                    @Override
                    public void flatMap(String s, Collector<TrafficVistorBean> collector) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(s);
                        JSONObject common = jsonObject.getJSONObject("common");
                        JSONObject page = jsonObject.getJSONObject("page");
                        long sv = 0L;
                        if (page.getString("last_page_id") == null) {
                            sv = 1L;
                        }
                        collector.collect(TrafficVistorBean.builder()
                                .svCt(sv)
                                .pvCt(1L)
                                .uvCt(0L)
                                .ujCt(0L)
                                .isNew(common.getString("is_new"))
                                .durSum(page.getLong("during_time"))
                                .ts(jsonObject.getLong("ts"))
                                .build());
                    }
                });
        SingleOutputStreamOperator<TrafficVistorBean> ujBeanStream =
                ujStream.flatMap(new FlatMapFunction<String, TrafficVistorBean>() {
                    @Override
                    public void flatMap(String s, Collector<TrafficVistorBean> collector) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(s);
                        JSONObject common = jsonObject.getJSONObject("common");
                        collector.collect(TrafficVistorBean.builder()
                                .ujCt(1L)
                                .uvCt(0L)
                                .svCt(0L)
                                .pvCt(0L)
                                .durSum(0L)
                                .isNew(common.getString("is_new"))
                                .ts(jsonObject.getLong("ts"))
                                .build());
                    }
                });
        //TODO 5 union连接3条流的数据
        DataStream<TrafficVistorBean> unionStream = ujBeanStream.union(pageBeanStream).union(uvBeanStream);
        //TODO 6 设置水位线
        SingleOutputStreamOperator<TrafficVistorBean> beanWithWatermarkStream =
                unionStream.assignTimestampsAndWatermarks(WatermarkStrategy.<TrafficVistorBean>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<TrafficVistorBean>() {
                    @Override
                    public long extractTimestamp(TrafficVistorBean trafficVistorBean, long l) {
                        return trafficVistorBean.getTs();
                    }
                }));
        //TODO 7 开窗聚合数据
        KeyedStream<TrafficVistorBean, String> keyedStream =
                beanWithWatermarkStream.keyBy(new KeySelector<TrafficVistorBean, String>() {
                    @Override
                    public String getKey(TrafficVistorBean trafficVistorBean) throws Exception {
                        return trafficVistorBean.getIsNew();
                    }
                });
        WindowedStream<TrafficVistorBean, String, TimeWindow> windowStream =
                keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(10)));
        SingleOutputStreamOperator<TrafficVistorBean> resultStream = windowStream.apply(new WindowFunction<TrafficVistorBean
                , TrafficVistorBean, String, TimeWindow>() {
            @Override
            public void apply(String s, TimeWindow timeWindow, Iterable<TrafficVistorBean> iterable,
                              Collector<TrafficVistorBean> collector) throws Exception {
                //独立访客数
                long uvct = 0L;
                //会话数
                long svCt = 0L;
                //页面浏览数
                long pvCt = 0L;
                //累计访问时长
                long durSum = 0L;
                //跳出会话数
                long ujCt = 0L;
                for (TrafficVistorBean trafficVistorBean : iterable) {
                    uvct += trafficVistorBean.getUvCt();
                    svCt += trafficVistorBean.getSvCt();
                    pvCt += trafficVistorBean.getPvCt();
                    durSum += trafficVistorBean.getDurSum();
                    ujCt += trafficVistorBean.getUjCt();
                }
                //保留两位小数
               // DecimalFormat df = new DecimalFormat("#.00");
                //会话平均浏览页数
                Double avgSessViPage = (double) pvCt / (double) svCt;
                //会话平均停留时长
                Double avgTime = (double) durSum / (double) svCt;
                //跳出率
                Double BounRate = (double) ujCt / (double) svCt;

                TrafficVistorBean trafficVistorBean = iterable.iterator().next();
                trafficVistorBean.setStt(DateFormatUtil.toYmdHms(timeWindow.getStart()));
                trafficVistorBean.setEdt(DateFormatUtil.toYmdHms(timeWindow.getEnd()));
                trafficVistorBean.setUvCt(uvct);
                trafficVistorBean.setSvCt(svCt);
                trafficVistorBean.setPvCt(pvCt);
                trafficVistorBean.setAvgSessViPage(new Double(avgSessViPage).longValue());
                trafficVistorBean.setDurSum(durSum);
                trafficVistorBean.setAvgTime(new Double(avgTime).longValue());
                trafficVistorBean.setUjCt(ujCt);
                trafficVistorBean.setBounRate(new Double(BounRate).longValue());
                trafficVistorBean.setTs(System.currentTimeMillis());
                collector.collect(trafficVistorBean);
            }
        });

        //TODO 8 写入到ClickHouse中
        resultStream.addSink(ClickHouseUtil.getClickHouseSinkFunc("insert into dws_traffic_visitor_window values (?," +
                "?,?,?,?,?,?,?,?,?,?,?)"));
        //TODO 9 执行任务
        env.execute(groupId);
    }
}